# 1、创建环境
from pyspark import StorageLevel
from pyspark.context import SparkContext

sc = SparkContext(master='local', appName='cache')

# 2、读取数据
lines_rdd = sc.textFile("../../data/students.txt")


def map_fun(line):
    split = line.split(",")
    id = split[0]
    name = split[1]
    age = split[2]
    sex = split[3]
    clazz = split[4]
    print("map_fun被执行")
    return id, name, age, sex, clazz


# 解析数据
students_rdd = lines_rdd.map(map_fun)

# 当同一个RDD被多次使用时，可以将RDD的数据缓存起来，缓存之后就不需要重复读取文件
# 当RDD被缓存之后，不会计算多次
# 缓存会将RDD的数据缓存在Executor中， 如果缓存级别是MEMORY_ONLY，RDD的数据不能超过了Executor内存的上限

# 默认缓存级别时MEMORY_ONLY
# students_rdd.cache()

# MEMORY_AND_DISK:内存放不下方磁盘，同时会对数据进行压缩
students_rdd.persist(StorageLevel.MEMORY_AND_DISK)

# 1、统计每个班级的人数
students_rdd \
    .map(lambda stu: (stu[-1], 1)) \
    .reduceByKey(lambda a, b: a + b) \
    .foreach(print)

# 2、统计每个性别的人数
students_rdd \
    .map(lambda stu: (stu[3], 1)) \
    .reduceByKey(lambda a, b: a + b) \
    .foreach(print)

# 2、统计每个年龄的人数
students_rdd \
    .map(lambda stu: (stu[2], 1)) \
    .reduceByKey(lambda a, b: a + b) \
    .foreach(print)

# 清除缓存
students_rdd.unpersist()

while True:
    pass
